package com.neusoft.databus.core.hbase;

import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.NavigableMap;

import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.data.hadoop.hbase.HbaseTemplate;
import org.springframework.stereotype.Component;

import com.neusoft.bizcore.web.utils.JsonUtils;
import com.neusoft.databus.common.bean.CommandBean;
import com.neusoft.databus.common.dto.MonitorDataDTO;
import com.neusoft.databus.common.dto.ResourceDTO;
import com.neusoft.databus.core.service.ResourceService;

/**
 * description
 *
 * @author sunchf
 * @date 2019年3月28日 下午4:53:25
 */
@Component
@ConditionalOnClass(HbaseTemplate.class)
@ConditionalOnProperty(name = "databus.core.hbase.enabled", havingValue = "true")
public class MonitorDataHBaseConverter {

    @Autowired
    private ResourceService resourceService;

    public static byte[] b_family = Bytes.toBytes("f");
    public static byte[] b_collectedTime = Bytes.toBytes("collectedTime");
    public static byte[] b_resource = Bytes.toBytes("resource");

    public MonitorDataDTO toDTO(final Result result) {
        final MonitorDataDTO dto = new MonitorDataDTO();
        dto.setResourceCode(Bytes
                .toString(result.getValue(MonitorDataHBaseConverter.b_family, MonitorDataHBaseConverter.b_resource)));
        try {
            final ResourceDTO resource = this.resourceService.findByCode(dto.getResourceCode());
            dto.setResourceName(resource.getName());
        } catch (final Exception e) {

        }
        dto.setCollectedTime(new Date(Bytes.toLong(
                result.getValue(MonitorDataHBaseConverter.b_family, MonitorDataHBaseConverter.b_collectedTime))));

        final NavigableMap<byte[], byte[]> navigableMap = result.getFamilyMap(MonitorDataHBaseConverter.b_family);
        final Map<String, String> map = new HashMap<>();
        navigableMap.entrySet().stream().forEach(entry -> {
            final String metricCode = Bytes.toString(entry.getKey());
            if (!("collectedTime".equals(metricCode) || "resource".equals(metricCode))) {
                map.put(metricCode, Bytes.toString(entry.getValue()));
            }
        });

        dto.setData(JsonUtils.pojoToJson(map));

        return dto;
    }

    public List<Put> toPut(final CommandBean commandBean) {

        final List<Put> list = new ArrayList<>();
        commandBean.getResourceBean().forEach(it -> {
            if (it.getMetrics().stream().filter(m -> m.isParseSuccess()).count() > 0) {
                final Put p = new Put(
                        Bytes.toBytes(this.getRowKey(it.getCode(), commandBean.getCollectEndTime().getTime())));
                p.addColumn(MonitorDataHBaseConverter.b_family, MonitorDataHBaseConverter.b_resource,
                        Bytes.toBytes(it.getCode()));
                p.addColumn(MonitorDataHBaseConverter.b_family, MonitorDataHBaseConverter.b_collectedTime,
                        Bytes.toBytes(commandBean.getCollectEndTime().getTime()));
                it.getMetrics().stream().filter(m -> m.isParseSuccess()).forEach(m -> {
                    p.addColumn(MonitorDataHBaseConverter.b_family, Bytes.toBytes(m.getCode()),
                            Bytes.toBytes(String.valueOf(m.getValue())));
                });

                list.add(p);
            }
        });

        return list;
    }

    private String getRowKey(String resourceCode, long collectEndTime) {
        final StringBuilder sb = new StringBuilder();
        sb.append(resourceCode);
        sb.append(collectEndTime);
        return sb.toString();
    }
}
